package fq.status;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;

import java.util.Map;
import java.util.Optional;

public class MyMapFunction  extends RichMapFunction<String,String> {

    private transient MapState<String,Integer> count; // 用于保存状态的变量，需要声明为transient以避免序列化问题

    @Override
    public void open(Configuration parameters) { // 在这里初始化状态
        MapStateDescriptor<String,Integer> descriptor = new MapStateDescriptor<>(
                "count", String.class, Integer.class); // 定义状态描述符，包括名称和类型
        count = getRuntimeContext().getMapState(descriptor); // 获取状态句柄
    }

    @Override
    public String map(String s) throws Exception {
        count.put(s, Optional.ofNullable(count.get(s)).orElse(0) +1);
        for (Map.Entry<String, Integer> entry : count.entries()) {
            System.out.println(entry.getKey() +": "+ entry.getValue());
        }
        System.out.println("-----------------" + count +" " + Thread.currentThread().getId());
        return s;
    }
}
